package org.xhy.hrm.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.xhy.hrm.constants.MQConstants;
import org.xhy.hrm.domain.KillOrder;
import org.xhy.hrm.domain.KillOrderItem;
import org.xhy.hrm.service.IKillOrderItemService;
import org.xhy.hrm.service.IKillOrderService;
import org.xhy.hrm.to.PayResultTo;

import java.io.IOException;
import java.util.Date;

import static org.xhy.hrm.constants.MQConstants.*;

@Component
@Slf4j
public class MessageConsumer {

    @Autowired
    private IKillOrderService killOrderService;

    @Autowired
    private IKillOrderItemService killOrderItemService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private RedisTemplate<Object,Object> redisTemplate;

    //下单处理
    @RabbitListener(queues = NAME_QUEUE_KILL_ORDER,containerFactory = "rabbitListenerContainerFactory")
    public void handleOrderMesssage(@Payload KillOrder killOrder, Message message, Channel channel) throws IOException {
        log.info("消费者的消息:->{}",killOrder);

        //分布式锁，防止集群创建多次订单
        RLock lock = redissonClient.getLock("LOCKORDER:" + killOrder.getOrderNo());
        lock.lock();
        try {
            //幂等处理
            //查询数据库订单
            KillOrder orderFromDB = killOrderService.selectByOrderNo(killOrder.getOrderNo());
            if (orderFromDB == null){
                //保存订单到数据库
                killOrder.setCreateTime(new Date());
                killOrder.setStatusOrder(KillOrder.STATUS_WAIT_PAY);
                killOrder.setLastUpdateTime(new Date());
                killOrderService.insert(killOrder);

                //订单明细
                KillOrderItem item = killOrder.getItem();
                item.setKillOrderId(killOrder.getId());
                item.setPrice(killOrder.getTotalPrice());
                item.setCount(killOrder.getKillCount());
                item.setCreateTime(killOrder.getCreateTime());
                killOrderItemService.insert(item);

                //下单成功后发送消息给延迟队列
                rabbitTemplate.convertAndSend(
                        MQConstants.NAME_EXCHANGE_KILL_DIE_TOPIC,
                        "killcoursedelay.order",
                        killOrder
                );

                //把订单放入一份到Redis用来支付
                redisTemplate.opsForValue().set("order:"+killOrder.getOrderNo(),killOrder);
            }
        } finally {
            if (lock.isLocked()){
                lock.unlock();
            }
        }

        //签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

    //支付过期，死信队列消费消息
    //消息过期，处理支付超时
    @RabbitListener(queues = NAME_QUEUE_KILL_DIE_ORDER , containerFactory = "rabbitListenerContainerFactory")
    public void handleDieOrderMesssage(@Payload KillOrder killOrder , Message message , Channel channel) throws IOException {
        log.info("过期消息:->{}",killOrder);

        RLock lock = redissonClient.getLock("LOCKORDERDIE:" + killOrder.getOrderNo());
        lock.lock();
        try {
            //查询数据库订单
            KillOrder orderFromDB = killOrderService.selectByOrderNo(killOrder.getOrderNo());
            //未支付，推库存
            if (orderFromDB != null && orderFromDB.getStatusOrder() == KillOrder.STATUS_WAIT_PAY){
                //未改状态为未支付
                orderFromDB.setStatusOrder(KillOrder.STATUS_PAY_FAIL);
                orderFromDB.setLastUpdateTime(new Date());
                killOrderService.updateById(orderFromDB);
                //信号量加库存
                Long killCourseId = killOrder.getItem().getKillCourseId();
                RSemaphore semaphore = redissonClient.getSemaphore("KILL_STORE:" + killCourseId.toString());
                semaphore.addPermits(1);

                //删除秒杀记录
                redisTemplate.delete("KILLLOG:"+killOrder.getUserId()+":"+killCourseId.toString());
            }
        } finally {
            if (lock.isLocked()){
                lock.unlock();
            }
        }
        //签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }


    //消费支付结果
    @RabbitListener(queues = NAME_QUEUE_KILL_PAY_RESULT , containerFactory = "rabbitListenerContainerFactory")
    public void handlePayResult(@Payload PayResultTo payResultTo , Message message , Channel channel) throws IOException {
        log.info("支付结果消息:->{}",payResultTo);

        RLock lock = redissonClient.getLock("LOCKPAYRESULT:" + payResultTo.getOrderNo());
        lock.lock();
        try {
            //查询数据库订单
            KillOrder orderFromDB = killOrderService.selectByOrderNo(payResultTo.getOrderNo());

            if (orderFromDB != null && orderFromDB.getStatusOrder() == KillOrder.STATUS_WAIT_PAY){
                //修改订单状态
                if (payResultTo.isSuccess()){
                    orderFromDB.setStatusOrder(KillOrder.STATUS_PAY_SUCCESS);
                } else {
                    orderFromDB.setStatusOrder(KillOrder.STATUS_PAY_FAIL);
                }
                orderFromDB.setLastUpdateTime(new Date());
                killOrderService.updateById(orderFromDB);
                //发货，拿到用户id和课程id，保存到一张新的表
                Long userId = orderFromDB.getUserId();
                Long courseId = orderFromDB.getItem().getKillCourseId();
                //调用Feign,保存用户课程购买记录到课程

                //删除秒杀记录
                redisTemplate.delete("KILLLOG:"+userId+":"+courseId.toString());
            }
        } finally {
            if (lock.isLocked()){
                lock.unlock();
            }
        }
        //签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

}
